package aggregation

import (
	context "context"
	"errors"
	"log"
	"net"
	"time"

	"gitee.com/bjf-fhe/apicat/entry"
	"gitee.com/bjf-fhe/apicat/entry/handler"
	"github.com/sirupsen/logrus"
	"google.golang.org/grpc"
	emptypb "google.golang.org/protobuf/types/known/emptypb"
)

type Server struct {
	ApicatRemoteAggregationServiceServer
	c         chan *entry.LogEntryResponse
	baseCtx   context.Context
	clients   map[string]*ClientStub
	judgement *handler.Judgement
}

type ClientStub struct {
	client        ApicatRemoteAggregationService_OrderServer
	OrderCancelFn context.CancelFunc
}

func (c *ClientStub) SendOrder(cmd *Command) error {
	var errC = make(chan error)
	go func() {
		errC <- c.client.Send(cmd)
	}()
	select {
	case err := <-errC:
		if err != nil {
			c.OrderCancelFn()
		}
		return err
	case <-time.After(5 * time.Second):
		c.OrderCancelFn()
		return errors.New("timeout")
	}
}

//处理完之后，通知下游
func (s *Server) RecordEntry(en *entry.LogEntry) error {
	if s.judgement.Judge(en) {
		if client, ok := s.clients[en.MachineId]; ok {
			return client.SendOrder(&Command{
				Type:  Type_TypeAddEntry,
				Entry: ToRemoteEntry(en),
			})
		}
		return errors.New("not found client")
	}
	return nil
}

// func (s *Server) NotifyAdd(mid string, en *entry.LogEntry) error {
// 	if client, ok := s.clients[mid]; ok {
// 		return client.SendOrder(&Command{
// 			Type:  Type_TypeAddEntry,
// 			Entry: ToRemoteEntry(en),
// 		})
// 	}
// 	return errors.New("not found client")
// }

// func (s *Server) NotifyDel(mid string, ens ...*entry.LogEntry) error {
// 	for _, en := range ens {
// 		if client, ok := s.clients[mid]; ok {
// 			err := client.SendOrder(&Command{
// 				Type:  Type_TypeDelEntry,
// 				Entry: ToRemoteEntry(en),
// 			})
// 			if err != nil {
// 				return err
// 			}
// 		} else {
// 			return errors.New("not found client")
// 		}
// 	}
// 	return nil
// }

func (s *Server) Report(ctx context.Context, remote *Entry) (*emptypb.Empty, error) {
	logrus.WithField("report", remote).Info("got report")
	s.c <- remote.ToLogEntryResponse()
	return &emptypb.Empty{}, nil
}

//下行通道
func (s *Server) Order(mi *MachineInfo, server ApicatRemoteAggregationService_OrderServer) error {
	//register in db
	ctx, fn := context.WithCancel(s.baseCtx)
	//first call
	var client *ClientStub
	var ok bool
	if client, ok = s.clients[mi.Id]; !ok {
		client = new(ClientStub)
		s.clients[mi.Id] = client
	} else if client != nil && client.client != server {
		logrus.Infoln("cancel the previous service")
		client.OrderCancelFn()
	} else if client.client != nil { //某种原因，同一个
		fn()
		return nil
	}
	client.client = server
	client.OrderCancelFn = fn
	<-ctx.Done()
	logrus.Infoln("stop Order", mi.Id)
	return nil
}

func (s *Server) Records() chan *entry.LogEntryResponse {
	logrus.Info("aggregate server start")
	return s.c
}

//main 从sub获取数据
func NewServer(addr string, judement *handler.Judgement, ctx context.Context) (server *Server, err error) {
	//创建grpc服务器
	var lis net.Listener
	lis, err = net.Listen("tcp", addr)
	if err == nil {
		s := grpc.NewServer()
		server = &Server{
			c:         make(chan *entry.LogEntryResponse, 10),
			clients:   make(map[string]*ClientStub),
			baseCtx:   ctx,
			judgement: judement,
		}

		RegisterApicatRemoteAggregationServiceServer(s, server)
		go func() {
			err := s.Serve(lis)
			if err != nil {
				log.Fatal(err)
			}
		}()
		go func() {
			<-ctx.Done()
			s.GracefulStop()
		}()
	}
	return
}
